use super::{
    column_datas::{add_block_column, ColumnData, DbValue}, inrc_join_table::IncrJoinTable, strconv::to_db_value,
    tmp_table::TmpTable,
};
use crate::{
    base::{
        exec_sqls, get_ch_pool, get_columns, get_config, get_join_conf_by_file_name,
        load_join_table_data, reload_dicts, ImportType, JoinTable, SourceType, Table,
    },
    big_table::BigTable,
    calc::{get_conv_func, parse_calc_exp},
    data_proc::{get_divi_cols, AcblProcess, DayTrlsProc, MonthTrlsProc},
};
use chrono::NaiveDate;
use clickhouse_rs::{
    types::SqlType,
    Block,
};
pub struct DataTable<'a> {
    source: SourceType,
    columns: Vec<(String, SqlType)>,
    conf: &'a Table,
    line_datas: Vec<String>,
    tmp_table: Option<TmpTable>,
    inrc_join_table: Option<IncrJoinTable<'a>>,
    join_tables: Vec<&'a JoinTable>,
    column_separator: char,
    batch_count: usize,
}
impl<'a> DataTable<'a> {
    pub async fn from(
        source: SourceType,
        conf: &'a Table,
        column_separator: char,
    ) -> anyhow::Result<Self> {
        let columns = get_columns(&conf.table_name).await?;
        let mut tmp_table: Option<TmpTable> = None;
        if conf.db_calc_cols.len() > 0 || conf.acbl_procs.is_some() {
            let tmp_name = format!("tmp_{}", conf.file_name);
            let tmp = TmpTable::from(tmp_name);
            tmp.create(&columns).await?;
            tmp_table = Some(tmp)
        }
        let join_tables = get_join_conf_by_file_name(conf.file_name.as_str());
        let inrc_join_table = IncrJoinTable::new(conf);
        let batch_count=get_config().write_batch_count as usize;
        Ok(Self {
            source,
            columns,
            conf,
            line_datas: Vec::with_capacity(batch_count),
            tmp_table,
            inrc_join_table,
            join_tables,
            column_separator,
            batch_count
        })
    }
    pub async fn before_import(&self) -> anyhow::Result<()> {
        if self.conf.import_type == ImportType::TruncateAndInsert {
            let mut client = get_ch_pool().get_handle().await?;
            client
                .execute(format!("TRUNCATE TABLE {}", self.conf.table_name))
                .await?;
        }
        Ok(())
    }
    pub async fn insert_rows(&mut self, line_datas: Vec<String>) -> anyhow::Result<()> {
        self.line_datas.extend(line_datas);
        if self.line_datas.len() >= self.batch_count {
            self.save_buffer()
                .await
                .or_else(|err| Err(anyhow::anyhow!("save_buffer出错{err}")))?;
        }
        Ok(())
    }
    pub async fn flush(&mut self) -> anyhow::Result<()> {
        if self.line_datas.len() > 0 {
            self.save_buffer().await?;
        }
        Ok(())
    }
    pub async fn after_import(&self, date: &NaiveDate) -> anyhow::Result<()> {
        let date_str = date.format("%Y-%m-%d").to_string();
        if let Some(before_sqls) = self.conf.before_sqls.as_ref() {
            exec_sqls(before_sqls, &["{date}"], &[date_str.as_str()]).await?;
        }
        if let Some(inrc_join_table) = &self.inrc_join_table {
            let wc = if self.conf.import_type == ImportType::MonthTrls {
                let (date_col, _) = self
                    .columns
                    .iter()
                    .find(|(_, st)| st == &SqlType::Date)
                    .unwrap();
                Some(format!(
                    "{date_col}>='{}' and {date_col}<'{}'",
                    date.format("%Y-%m-01").to_string(),
                    date_str
                ))
            } else {
                None
            };
            inrc_join_table.create(wc.as_deref()).await?;
        }
        if let Some(tmp_table) = self.tmp_table.as_ref() {
            let mut divi_with: Option<String> = None;
            let mut divi_col_exps: Option<String> = None;
            if self.conf.divi_fact_convs.len() > 0 {
                if self.conf.import_type == ImportType::MonthTrls {
                    MonthTrlsProc::new(&self.conf, &self.columns, &tmp_table.name)
                        .run(date)
                        .await?;
                } else if self.conf.import_type == ImportType::DayTrls {
                    DayTrlsProc::new(&self.conf, &self.columns)
                        .run(date)
                        .await?;
                }
                if self.conf.acbl_procs.is_none() {
                    let (with_str, sel_cols) =
                        get_divi_cols(&self.conf.divi_fact_convs, date, None).await?;
                    divi_with = Some(with_str);
                    divi_col_exps = Some(sel_cols);
                }
            }
            self.insert_from_tmp(date, divi_with, divi_col_exps).await?;
            self.acco_close(date).await?;
        }

        if let Some(after_sqls) = self.conf.after_sqls.as_ref() {
            exec_sqls(after_sqls, &["{date}"], &[date_str.as_str()]).await?;
        }
        if let Some(inrc_join_table) = self.inrc_join_table.as_ref() {
            inrc_join_table.drop().await?;
        }
        for jt in self.join_tables.iter() {
            load_join_table_data(jt).await?;
        }
        let dicts = &self.conf.reload_dict_tables;
        reload_dicts(dicts).await?;
        if self.conf.import_type == ImportType::InsertAndUpdate && self.conf.acbl_procs.is_some() {
            AcblProcess::new(&self.conf).run(date).await?;
        }
        let bt = BigTable::new(self.source, &self.conf.file_name);
        bt.clear(date).await?;
        bt.build(date).await?;
        Ok(())
    }
    async fn insert_from_tmp(
        &self,
        date: &NaiveDate,
        divi_with: Option<String>,
        divi_col_exps: Option<String>,
    ) -> anyhow::Result<()> {
        if self.tmp_table.is_none() {
            return Ok(());
        }
        let tmp_table = self.tmp_table.as_ref().unwrap();
        let mut ins_col_names = String::new();
        let mut sel_col_names = String::new();
        for (cname, _) in self.columns.iter() {
            if cname.starts_with("DIVI.") || cname == "SIGN" {
                continue;
            }
            ins_col_names.push_str(cname);
            let ccol = self.conf.db_calc_cols.iter().find(|(k, _)| *k == cname);
            if let Some((_k, v)) = ccol {
                if v == "#CLDT" {
                    sel_col_names.push_str("toDate('1970-01-01')");
                } else if v == "#IPDT" {
                    sel_col_names.push_str("now()");
                } else if v == "#INCR" {
                    let ijt = self.inrc_join_table.as_ref().unwrap();
                    let key_exp = if self.conf.unique_keys.len() == 1 {
                        self.conf.unique_keys[0].clone()
                    } else {
                        format!("concat({})", self.conf.unique_keys.join(","))
                    };
                    let col_exp = format!("joinGet('{}','ID',{key_exp})", ijt.get_name());
                    sel_col_names.push_str(&col_exp);
                } else if v == "#DATE" {
                    let date_str = format!("toDate('{}')", date.format("%Y-%m-%d").to_string());
                    sel_col_names.push_str(date_str.as_str());
                } else {
                    sel_col_names.push_str(v);
                }
            } else {
                sel_col_names.push_str(cname);
            }
            ins_col_names.push_str(",");
            sel_col_names.push_str(",");
        }
        ins_col_names.pop();
        sel_col_names.pop();
        if let Some(ijt) = self.inrc_join_table.as_ref() {
            ijt.insert_new().await?;
        }
        let with_cause = if let Some(wc) = self.conf.with_cause.as_ref() {
            if let Some(dw) = divi_with {
                format!("WITH {wc},{dw}")
            } else {
                format!("WITH {wc}")
            }
        } else {
            if let Some(dw) = divi_with {
                format!("WITH {dw}")
            } else {
                "".to_owned()
            }
        };
        let mut where_cause = String::new();
        if self.conf.import_type == ImportType::MonthTrls {
            let clen = self.conf.acbl_col_names.len();
            for (i, c) in self.conf.acbl_col_names.iter().enumerate() {
                let exp = format!("{}!=0", c);
                where_cause.push_str(&exp);
                if i < clen - 1 {
                    where_cause.push_str(" OR ");
                }
            }
        }
        if !where_cause.is_empty() {
            where_cause = format!("where {where_cause}")
        }
        let sql_str = if let Some(dces) = divi_col_exps {
            format!(
                "insert into {}({ins_col_names},DIVI.DPID,DIVI.EMID,DIVI.DSVL,SIGN) {with_cause} select {sel_col_names},{dces},1 from {} {where_cause}",
                self.conf.table_name, tmp_table.name
            )
        } else {
            format!(
                "insert into {}({ins_col_names}) {with_cause} select {sel_col_names} from {} {where_cause}",
                self.conf.table_name, tmp_table.name
            )
        };
        let mut client = get_ch_pool().get_handle().await?;
        client.execute(&sql_str).await?;
        Ok(())
    }
    async fn acco_close(&self, date: &NaiveDate) -> anyhow::Result<()> {
        if self.conf.acbl_procs.is_none() {
            return Ok(());
        }
        let tmp_table = self.tmp_table.as_ref().unwrap();
        let mut ins_col_names = String::new();
        let mut sel_col_names = String::new();
        for (i, (cname, _)) in self.columns.iter().enumerate() {
            ins_col_names.push_str(cname);
            if self.conf.acbl_col_names.iter().any(|c| c == cname) {
                sel_col_names.push_str("0");
            } else {
                let ccol = self.conf.db_calc_cols.iter().find(|(k, _)| *k == cname);
                if let Some((_, v)) = ccol {
                    if v == "#CLDT" {
                        let date_str = format!("toDate('{}')", date.format("%Y-%m-%d").to_string());
                        sel_col_names.push_str(&date_str);
                    } else if v == "#IPDT" {
                        sel_col_names.push_str("now()");
                    } else {
                        sel_col_names.push_str(cname);
                    }
                } else {
                    sel_col_names.push_str(cname);
                }
            }
            if i < self.columns.len() - 1 {
                ins_col_names.push_str(",");
                sel_col_names.push_str(",");
            }
        }
        let ukey_str = if self.conf.unique_keys.len() == 1 {
            self.conf.unique_keys[0].clone()
        } else {
            format!("concat({})", self.conf.unique_keys.join(","))
        };
        let sql_str = format!(
            "insert into {}({}) select {} from {} final where {ukey_str} not in (select {ukey_str} from {})",
            self.conf.table_name,ins_col_names,sel_col_names,self.conf.table_name,tmp_table.get_name()
        );
        let mut client = get_ch_pool().get_handle().await?;
        client.execute(sql_str).await?;
        Ok(())
    }
    fn exec_rs_calc(&self, row_data: &Vec<&str>) -> anyhow::Result<Vec<(&str, String)>> {
        let mut calcs: Vec<(&str, String)> = Vec::new();
        for (c, cf) in self.conf.rs_calc_cols.iter() {
            if !cf.contains('(') && !cf.contains(')') {
                calcs.push((c, cf.to_string()));
                continue;
            }
            let (fname, paras) = parse_calc_exp(cf)?;
            let mut para_vals: Vec<&str> = Vec::new();
            for p in paras.iter() {
                if p.starts_with("'") {
                    if p.ends_with("'") {
                        let v = &p[1..p.len() - 1];
                        para_vals.push(v);
                    } else {
                        return Err(anyhow::anyhow!(
                            "rs_calc_cols:列{}配置错误，参数 {} 以'开头，没有以'结尾",
                            c,
                            p
                        ));
                    }
                } else {
                    if p.as_bytes().iter().all(|b| b.is_ascii_digit()) {
                        para_vals.push(p.as_str());
                        continue;
                    }
                    if let Some((_, v)) = calcs.iter().find(|(k, _)| k == p) {
                        para_vals.push(v);
                    } else {
                        let i = self.conf.columns.get(p).ok_or(anyhow::anyhow!(
                            "rs_calc_cols:列{}配置错误，列 {} 在表 {} 中不存在",
                            c,
                            p,
                            self.conf.table_name
                        ))?;
                        let idx2 = *i as usize;
                        let val = row_data.get(idx2).ok_or(anyhow::anyhow!(
                            "列{}配置对应文本索引{}超出范围",
                            c,
                            idx2
                        ))?;
                        para_vals.push(val);
                    }
                }
            }
            if let Some(calc_fn) = get_conv_func(&fname) {
                let v = calc_fn(para_vals)?;
                calcs.push((c, v));
            } else {
                return Err(anyhow::anyhow!(
                    "rs_calc_cols:列{}配置错误，函数 {} 未找到",
                    c,
                    fname
                ));
            }
        }
        Ok(calcs)
    }
    async fn save_buffer(&mut self) -> anyhow::Result<()> {
        let mut client = get_ch_pool().get_handle().await?;
        let tname = if let Some(tmp) = &self.tmp_table {
            tmp.get_name()
        } else {
            self.conf.table_name.as_str()
        };

        let calc_len = self.conf.rs_calc_cols.len();
        let mut block = Block::new();

        // 预处理所有行数据和计算列
        let (row_datas, calc_cols): (Vec<Vec<&str>>, Vec<Vec<(&str, String)>>) = self
            .line_datas
            .iter()
            .map(|line| {
                let row_data: Vec<&str> = line.split(self.column_separator).collect();
                let calcs = if calc_len > 0 {
                    self.exec_rs_calc(&row_data).unwrap_or_default()
                } else {
                    Vec::new()
                };
                (row_data, calcs)
            })
            .unzip();

        // 构建列数据并填充到 Block
        for (name, st) in self.columns.iter() {
            let mut db_vals: Vec<DbValue> = Vec::with_capacity(self.line_datas.len());
            if name == "SIGN" {
                db_vals.resize(self.line_datas.len(), DbValue::Int8(1));
            } else if let Some(idx) = self.conf.columns.get(name) {
                // 直接从行数据中提取
                for row in &row_datas {
                    if let Some(val) = row.get(*idx as usize) {
                        db_vals.push(to_db_value(val, st)?);
                    }
                }
            } else if calc_len > 0 {
                // 从计算列中提取
                for calcs in &calc_cols {
                    if let Some((_, val)) = calcs.iter().find(|(k, _)| k == name) {
                        db_vals.push(to_db_value(val, st)?);
                    }
                }
            }
            if !db_vals.is_empty() {
                let col_data = ColumnData::new(name, db_vals);
                block = add_block_column(block, st, col_data)?;
            }
        }
        client.insert(tname, block).await?;
        self.line_datas.clear();
        Ok(())
    }
}
